package com.gitee.xhs.broadcastState;

import com.gitee.xhs.sampleRecod.transactionRecord.Transaction;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import static com.gitee.xhs.broadcastState.BroadcastStateConstant.BROADCAST_STATE_MAP_STATE_DESCRIPTOR;

/**
 * @author BigHead
 * @date 2022/7/6 17:46
 * @description BroadcastStateBroadcastProcessFunction
 */
public class BroadcastStateBroadcastProcessFunction extends BroadcastProcessFunction<Transaction, BroadcastStateDao, BroadcastStateDao> {
    @Override
    public void processElement(Transaction value, BroadcastProcessFunction<Transaction, BroadcastStateDao, BroadcastStateDao>.ReadOnlyContext ctx, Collector<BroadcastStateDao> out) throws Exception {
        if (ctx.getBroadcastState(BROADCAST_STATE_MAP_STATE_DESCRIPTOR).contains(value.getAccountId())) {
            out.collect(
                    ctx.getBroadcastState(BROADCAST_STATE_MAP_STATE_DESCRIPTOR).get(value.getAccountId())
            );
        }
    }

    @Override
    public void processBroadcastElement(BroadcastStateDao value, BroadcastProcessFunction<Transaction, BroadcastStateDao, BroadcastStateDao>.Context ctx, Collector<BroadcastStateDao> out) throws Exception {
        ctx.getBroadcastState(BROADCAST_STATE_MAP_STATE_DESCRIPTOR).put(value.getAccountId(), value);
    }
}
